package com.shujia.base;

import com.shujia.utils.HbaseUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class HbaseFilter {
    //成员变量
    private Connection conn;
    private Admin admin;

    @Before
    public void connection() {
        try {
            //新版本创建配置文件对象的方式
            Configuration conf = HBaseConfiguration.create();

            //设置zookeeper的节点信息
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

            //创建hbase连接对象
            conn = ConnectionFactory.createConnection(conf);

            //创建数据库操作对象
            admin = conn.getAdmin();

            System.out.println("成功获取数据库连接对象：" + conn);
            System.out.println("成功获取数据库操作对象：" + admin);
            System.out.println("=================================================");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求1：通过RowFilter与BinaryComparator过滤比rowKey 1500100010小的所有值出来
     */
    @Test
    public void rowFilterWithBinaryComparator() {
        try {
            //将表名封装成一个TableName对象
            TableName tableName = TableName.valueOf("students");

            /**
             * 判断表是否存在
             */
            if (!admin.tableExists(tableName)) {
                System.out.println(Bytes.toString(tableName.getName()) + " 表不存在！无法查询数据！");
                return;
            }

            //获取表实例对象
            Table students = conn.getTable(tableName);
            //创建一个Scan对象
            Scan scan = new Scan(); //默认是查询所有行数据
            //创建BinaryComparator比较器
            //public BinaryComparator(byte[] value)
            BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("1500100010"));

            //创建行键过滤器对象
            //旧版本写法：public RowFilter(final CompareOp rowCompareOp, final ByteArrayComparable rowComparator)
//            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.LESS_OR_EQUAL, binaryComparator);
            //新版本写法：public RowFilter(final CompareOperator op, final ByteArrayComparable rowComparator)
            RowFilter rowFilter = new RowFilter(CompareOperator.LESS, binaryComparator);

            //public Scan setFilter(Filter filter)
            scan.setFilter(rowFilter);

            //表调用方法查询数据
            ResultScanner studentsScanner = students.getScanner(scan);
            Iterator<Result> resultIterator = studentsScanner.iterator();
            while (resultIterator.hasNext()) {
                Result result = resultIterator.next();
                HbaseUtil.printResult(result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求2：通过FamilyFilter与SubstringComparator查询列簇名包含in的所有列簇下面的数据
     */
    @Test
    public void familyFilterWithSubstringComparator() {
        try {
            //将表名封装成一个TableName对象
            TableName tableName = TableName.valueOf("users");

            /**
             * 判断表是否存在
             */
            if (!admin.tableExists(tableName)) {
                System.out.println(Bytes.toString(tableName.getName()) + " 表不存在！无法查询数据！");
                return;
            }

            //获取表实例对象
            Table students = conn.getTable(tableName);
            //创建一个Scan对象
            Scan scan = new Scan(); //默认是查询所有行数据

            //创建包含比较器
            SubstringComparator substringComparator = new SubstringComparator("in");

            //创建列簇过滤器
            //public FamilyFilter(final CompareOperator op,final ByteArrayComparable familyComparator)
            FamilyFilter familyFilter = new FamilyFilter(CompareOperator.EQUAL, substringComparator);

            scan.setFilter(familyFilter);

            //表调用方法查询数据
            ResultScanner studentsScanner = students.getScanner(scan);
            Iterator<Result> resultIterator = studentsScanner.iterator();
            while (resultIterator.hasNext()) {
                Result result = resultIterator.next();
                HbaseUtil.printResult(result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求3：通过QualifierFilter与SubstringComparator查询列名包含ge的列的值
     */
    @Test
    public void qualifierFilterWithSubstringComparator() {
        try {
            //将表名封装成一个TableName对象
            TableName tableName = TableName.valueOf("students");

            /**
             * 判断表是否存在
             */
            if (!admin.tableExists(tableName)) {
                System.out.println(Bytes.toString(tableName.getName()) + " 表不存在！无法查询数据！");
                return;
            }

            //获取表实例对象
            Table students = conn.getTable(tableName);
            //创建一个Scan对象
            Scan scan = new Scan(); //默认是查询所有行数据
            //创建包含比较器
            SubstringComparator substringComparator = new SubstringComparator("ge");

            //创建列名过滤器
            //注意：只会过滤出符合条件的列
            //public QualifierFilter(final CompareOperator op, final ByteArrayComparable qualifierComparator)
            QualifierFilter qualifierFilter = new QualifierFilter(CompareOperator.EQUAL, substringComparator);

            scan.setFilter(qualifierFilter);

            //表调用方法查询数据
            ResultScanner studentsScanner = students.getScanner(scan);
            Iterator<Result> resultIterator = studentsScanner.iterator();
            while (resultIterator.hasNext()) {
                Result result = resultIterator.next();
                HbaseUtil.printResult(result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求4：通过ValueFilter与BinaryPrefixComparator过滤出所有的cell中值以 "张" 开头的学生
     */
    @Test
    public void valueFilterWithBinaryPrefixComparator() {
        try {
            //将表名封装成一个TableName对象
            TableName tableName = TableName.valueOf("students");

            /**
             * 判断表是否存在
             */
            if (!admin.tableExists(tableName)) {
                System.out.println(Bytes.toString(tableName.getName()) + " 表不存在！无法查询数据！");
                return;
            }

            //获取表实例对象
            Table students = conn.getTable(tableName);
            //创建一个Scan对象
            Scan scan = new Scan(); //默认是查询所有行数据
            //创建二进制前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("文科"));

            //创建列值过滤器
            //public ValueFilter(final CompareOperator valueCompareOp, final ByteArrayComparable valueComparator)
            ValueFilter valueFilter = new ValueFilter(CompareOperator.EQUAL, binaryPrefixComparator);


            scan.setFilter(valueFilter);

            //表调用方法查询数据
            ResultScanner studentsScanner = students.getScanner(scan);
            Iterator<Result> resultIterator = studentsScanner.iterator();
            while (resultIterator.hasNext()) {
                Result result = resultIterator.next();
                HbaseUtil.printResult(result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 专用过滤器：单列值过滤器 返回的符合条件一行所有列数据
     * 需求1：通过SingleColumnValueFilter与查询文科班所有学生信息
     */
    @Test
    public void singleColumnValueFilterWithBinaryPrefixComparator() {
        try {
            //将表名封装成一个TableName对象
            TableName tableName = TableName.valueOf("students");

            /**
             * 判断表是否存在
             */
            if (!admin.tableExists(tableName)) {
                System.out.println(Bytes.toString(tableName.getName()) + " 表不存在！无法查询数据！");
                return;
            }

            //获取表实例对象
            Table students = conn.getTable(tableName);
            //创建一个Scan对象
            Scan scan = new Scan(); //默认是查询所有行数据
            //创建二进制前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("文科"));

            //创建单列值过滤器
            //public SingleColumnValueFilter(final byte [] family, final byte [] qualifier,
            //      final CompareOperator op,
            //      final org.apache.hadoop.hbase.filter.ByteArrayComparable comparator)
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"),
                    Bytes.toBytes("clazz"),
                    CompareOperator.EQUAL,
                    binaryPrefixComparator);


            scan.setFilter(singleColumnValueFilter);

            //表调用方法查询数据
            ResultScanner studentsScanner = students.getScanner(scan);
            Iterator<Result> resultIterator = studentsScanner.iterator();
            while (resultIterator.hasNext()) {
                Result result = resultIterator.next();
                HbaseUtil.printResult(result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求2：使用单列值排除过滤器
     * 通过SingleColumnValueExcludeFilter与BinaryComparator查询文科一班所有学生信息
     */
    @Test
    public void singleColumnValueExcludeFilterWithBinaryPrefixComparator() {
        try {
            //将表名封装成一个TableName对象
            TableName tableName = TableName.valueOf("students");

            /**
             * 判断表是否存在
             */
            if (!admin.tableExists(tableName)) {
                System.out.println(Bytes.toString(tableName.getName()) + " 表不存在！无法查询数据！");
                return;
            }

            //获取表实例对象
            Table students = conn.getTable(tableName);
            //创建一个Scan对象
            Scan scan = new Scan(); //默认是查询所有行数据
            //创建二进制前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("文科一班"));

            //创建单列值过滤器
            //public SingleColumnValueFilter(final byte [] family, final byte [] qualifier,
            //      final CompareOperator op,
            //      final org.apache.hadoop.hbase.filter.ByteArrayComparable comparator)
            SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(Bytes.toBytes("info"),
                    Bytes.toBytes("clazz"),
                    CompareOperator.EQUAL,
                    binaryPrefixComparator);


            scan.setFilter(singleColumnValueExcludeFilter);

            //表调用方法查询数据
            ResultScanner studentsScanner = students.getScanner(scan);
            Iterator<Result> resultIterator = studentsScanner.iterator();
            while (resultIterator.hasNext()) {
                Result result = resultIterator.next();
                HbaseUtil.printResult(result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求3：通过PrefixFilter查询以15001001开头的所有前缀的rowkey
     */
    @Test
    public void prefixFilter() {
        try {
            //将表名封装成一个TableName对象
            TableName tableName = TableName.valueOf("students");

            /**
             * 判断表是否存在
             */
            if (!admin.tableExists(tableName)) {
                System.out.println(Bytes.toString(tableName.getName()) + " 表不存在！无法查询数据！");
                return;
            }

            //获取表实例对象
            Table students = conn.getTable(tableName);
            //创建一个Scan对象
            Scan scan = new Scan(); //默认是查询所有行数据
            //创建前缀过滤器
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("150010002"));

            scan.setFilter(prefixFilter);

            //表调用方法查询数据
            ResultScanner studentsScanner = students.getScanner(scan);
            Iterator<Result> resultIterator = studentsScanner.iterator();
            while (resultIterator.hasNext()) {
                Result result = resultIterator.next();
                HbaseUtil.printResult(result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求：过滤出姓田，男生，年龄大于21，文科
     * 组合过滤器的使用
     */
    @Test
    public void groupFilter() {
        try {
            //将表名封装成一个TableName对象
            TableName tableName = TableName.valueOf("students");

            /**
             * 判断表是否存在
             */
            if (!admin.tableExists(tableName)) {
                System.out.println(Bytes.toString(tableName.getName()) + " 表不存在！无法查询数据！");
                return;
            }

            //获取表实例对象
            Table students = conn.getTable(tableName);
            //创建一个Scan对象
            Scan scan = new Scan(); //默认是查询所有行数据

            //创建一个存储Filter对象的集合
            ArrayList<Filter> filters = new ArrayList<>();

            //过滤出姓田
            //创建单列值过滤器
            SingleColumnValueFilter filter1 = new SingleColumnValueFilter(Bytes.toBytes("info"),
                    Bytes.toBytes("name"),
                    CompareOperator.EQUAL,
                    new BinaryPrefixComparator(Bytes.toBytes("丁")));
            filters.add(filter1);

            //男生
            SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes("info"),
                    Bytes.toBytes("gender"),
                    CompareOperator.EQUAL,
                    new BinaryComparator(Bytes.toBytes("男")));
            filters.add(filter2);

            //年龄大于21
            SingleColumnValueFilter filter3 = new SingleColumnValueFilter(Bytes.toBytes("info"),
                    Bytes.toBytes("age"),
                    CompareOperator.GREATER,
                    new BinaryComparator(Bytes.toBytes("21")));
            filters.add(filter3);


            //文科
            SingleColumnValueFilter filter4 = new SingleColumnValueFilter(Bytes.toBytes("info"),
                    Bytes.toBytes("clazz"),
                    CompareOperator.EQUAL,
                    new BinaryPrefixComparator(Bytes.toBytes("文科")));
            filters.add(filter4);


            //如果以一次设置一个过滤器来做的话，只有最后一个过滤器生效
//            scan.setFilter(filter1);
//            scan.setFilter(filter2);
//            scan.setFilter(filter3);
//            scan.setFilter(filter4);

            //如果要使用组合过滤器的话，使用FilterList类来实现
            FilterList filterList = new FilterList();
//            filterList.addFilter(filter1);
//            filterList.addFilter(filter2);
//            filterList.addFilter(filter3);
//            filterList.addFilter(filter4);
            filterList.addFilter(filters);


            scan.setFilter(filterList);

            //表调用方法查询数据
            ResultScanner studentsScanner = students.getScanner(scan);
            Iterator<Result> resultIterator = studentsScanner.iterator();
            while (resultIterator.hasNext()) {
                Result result = resultIterator.next();
                HbaseUtil.printResult(result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * PageFilter 分页过滤器
     * 需求：查询5页数据，每页10条
     */
    @Test
    public void PageFilter() {
        try {
            //将表名封装成TableName的对象
            TableName tableName = TableName.valueOf("students");

            //获取表
            Table students = conn.getTable(tableName);

            //定义一些变量
            //定义查询的页数
            int pageNumber = 5;

            //定义每一页的条数
            int pageDataNumber = 10;

            //设置最开始的行
            Scan scan = new Scan();
            scan.withStartRow(Bytes.toBytes(""));

            //创建一个分页过滤器
            PageFilter pageFilter = new PageFilter(pageDataNumber);
            scan.setFilter(pageFilter);

            for (int i = 1; i <= pageNumber;i++) {
                System.out.println("==================当前是第 "+i+" 页数据======================");
                ResultScanner resultScanner = students.getScanner(scan);

                Iterator<Result> iterator = resultScanner.iterator();
                while (iterator.hasNext()){
                    Result result = iterator.next();
                    /**
                     * 关键代码
                     */
                    //TODO: 关键代码，查找下一条数据的关键
                    scan.withStartRow(Bytes.toBytes(Bytes.toString(result.getRow())+"0"));
                    HbaseUtil.printResult(result);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * SkipFiler
     */
    @Test
    public void SkipFiler(){
        try {
            //将表名封装成TableName的对象
            TableName tableName = TableName.valueOf("students");

            //获取表
            Table students = conn.getTable(tableName);

            //创建列值过滤器
            ValueFilter valueFilter = new ValueFilter(CompareOperator.NOT_EQUAL, new BinaryComparator(Bytes.toBytes("易政")));

            //创建Skip过滤器
            SkipFilter skipFilter = new SkipFilter(valueFilter);


            //设置最开始的行
            Scan scan = new Scan();
            scan.setFilter(skipFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            Iterator<Result> iterator = resultScanner.iterator();
            while (iterator.hasNext()){
                Result result = iterator.next();

                HbaseUtil.printResult(result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * WhileMatchFilter
     */
    @Test
    public void WhileMatchFilter(){
        try {
            //将表名封装成TableName的对象
            TableName tableName = TableName.valueOf("students");

            //获取表
            Table students = conn.getTable(tableName);

            //创建列值过滤器
            ValueFilter valueFilter = new ValueFilter(CompareOperator.NOT_EQUAL, new BinaryComparator(Bytes.toBytes("易政")));

            //创建WhileMatchFilter过滤器
            WhileMatchFilter whileMatchFilter = new WhileMatchFilter(valueFilter);


            //设置最开始的行
            Scan scan = new Scan();
            scan.setFilter(whileMatchFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            Iterator<Result> iterator = resultScanner.iterator();
            while (iterator.hasNext()){
                Result result = iterator.next();

                HbaseUtil.printResult(result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @After
    public void close() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
